
package org.jmqtt.broker.store.redis.support;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.jmqtt.broker.common.config.BrokerConfig;
import org.jmqtt.broker.common.log.JmqttLogger;
import org.jmqtt.broker.common.log.LogUtil;
import org.jmqtt.broker.store.redis.RedisCallBack;
import org.slf4j.Logger;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisSentinelPool;

import java.util.HashSet;
import java.util.Set;

/**
 * redis哨兵支持实现
 *
 * @author nn200433
 * @date 2021-07-21 03:27:04
 */
public class RedisSentinelSupportImpl implements RedisSupport {

    private static final Logger log = JmqttLogger.storeLog;

    private BrokerConfig brokerConfig;
    private JedisSentinelPool jedisPool;

    public RedisSentinelSupportImpl(BrokerConfig brokerConfig) {
        this.brokerConfig = brokerConfig;
    }

    @Override
    public void init() {
        try {
            JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
            jedisPoolConfig.setMinIdle(brokerConfig.getMinIdle());
            jedisPoolConfig.setMaxTotal(brokerConfig.getMaxTotal());
            jedisPoolConfig.setTestOnBorrow(brokerConfig.isTestOnBorrow());
            jedisPoolConfig.setMaxTotal(jedisPoolConfig.getMaxIdle());
            jedisPoolConfig.setMaxWaitMillis(jedisPoolConfig.getMaxWaitMillis());

            String[] redisHosts = brokerConfig.getRedisHost().split(",");
            if (ArrayUtils.isEmpty(redisHosts)) {
                throw new Exception("array [redisHost] is empty");
            }

            Set<String> redisNodes = new HashSet<String>(redisHosts.length);
            for (String redisHost : redisHosts) {
                String[] ipPort = redisHost.split(":");
                if (ipPort.length != 2) {
                    throw new Exception("redisHost wrong format, example: 127.0.0.1:6379,127.0.0.1:6380");
                } else {
                    redisNodes.add(redisHost);
                }
            }

            if (StringUtils.isEmpty(brokerConfig.getRedisPassword())) {
                throw new Exception("redisPassword is empty");
            }

            jedisPool = new JedisSentinelPool(brokerConfig.getMasterName(), redisNodes,
                    jedisPoolConfig, 10000, brokerConfig.getRedisPassword(), brokerConfig.getRedisDataBase());

            LogUtil.debug(log, "[Redis handle] JedisSentinelPool init success");
        } catch (Exception ex) {
            LogUtil.error(log, "[Redis handle error],ex:{}", ex);
        }
    }

    @Override
    public <T> T operate(RedisCallBack redisCallBack) {
        LogUtil.debug(log, "[Cluster] redis operate begin");
        long startTime = System.currentTimeMillis();
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            return (T) redisCallBack.operate(jedis);
        } catch (Exception ex) {
            LogUtil.error(log, "[Cluster] redis operate error,ex:{}", ex);
        } finally {
            LogUtil.debug(log, "[Cluster] redis operate cost:{}", (System.currentTimeMillis() - startTime));
            if (jedis != null) {
                jedis.close();
            }
        }
        return null;
    }

    @Override
    public void close() {
        LogUtil.info(log, "[Cluster] redis close");
        if (jedisPool != null) {
            jedisPool.close();
        }
    }

}
